package com.own.business.sync.trigger.polling.core;

import com.own.business.sync.trigger.base.bak.util.SyncTriggerBakUtil;
import com.own.business.sync.trigger.base.config.SyncTriggerPollingCondition;
import com.own.business.sync.trigger.base.entity.SyncTrigger;
import com.own.business.sync.trigger.base.method.ResolveMethodSpringHook;
import com.own.business.sync.trigger.base.property.SyncTriggerProperty;
import com.own.business.sync.trigger.base.service.SyncTriggerService;
import com.own.component.common.util.executor.ExecutorUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.Conditional;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.function.BiConsumer;
import java.util.function.Consumer;

/**
 * SyncTriggerPollingCoreTask
 * <p>
 * 启动完成后开始执行
 *
 * @author chenxueli
 * @date 2024-01-08 15:48:43
 */
@Slf4j
@Component
@Conditional(SyncTriggerPollingCondition.class)
public class SyncTriggerPollingCoreTask {

    private final Object lock = new Object();
    @Resource
    private SyncTriggerProperty syncTriggerProperty;
    @Resource
    private SyncTriggerService syncTriggerService;
    @Resource
    private SyncTriggerBakUtil syncTriggerBakUtil;
    @Resource
    private ResolveMethodSpringHook resolveMethodSpringHook;
    @Resource
    private ExecutorUtil executorUtil;

    @PostConstruct
    public void init() {
        executorUtil.execute(() -> {
            if (syncTriggerProperty.getUseVirtualThread()) {
                // 创建虚拟线程
                var service = ExecutorUtil.newVirtualThreadPerTaskExecutor();
                run((operation, data) -> service.submit(() -> {
                    try {
                        operation.accept(data);
                    } catch (Exception e) {
                        log.debug("处理异常", e);
                        syncTriggerBakUtil.save(data, e.getMessage());
                    }
                }));
            } else {
                run((operation, data) -> {
                    try {
                        operation.accept(data);
                    } catch (Exception e) {
                        log.debug("处理异常", e);
                        syncTriggerBakUtil.save(data, e.getMessage());
                    }
                });
            }
        });
    }

    /**
     * 轮询信息
     *
     * @param runnable 具体执行的方法
     */
    @SuppressWarnings("all")
    private void run(BiConsumer<Consumer<SyncTrigger>, SyncTrigger> runnable) {
        var max = syncTriggerProperty.getPollingTimeList().length - 1;
        // 采用while true的方式可以避免使用递归方式造成的栈溢出的问题
        var c = 0;
        while (true) {
            synchronized (lock) {
                // 查询当前备份库的数量信息
                var count = syncTriggerService.count();
                // 如果大于0，开始轮询信息
                if (count > 0) {
                    log.info("查询到{}条数据需要处理", count);
                    for (var i = 0; i < count; i++) {
                        // 获取到最新的数据去执行信息
                        var data = syncTriggerService.getOldestData();
                        if (data == null) {
                            log.info("数据为空，跳过本次处理");
                            break;
                        }
                        var method = resolveMethodSpringHook.getMethod(data.getModuleName());
                        if (method != null) {
                            var operation = resolveMethodSpringHook.getMethod(data.getType(), method);
                            runnable.accept(operation, data);
                        } else {
                            syncTriggerBakUtil.save(data, "未找到处理方法");
                        }
                        syncTriggerService.deleteById(data.getId());
                        log.debug("处理完成:{},{},{}", data.getModuleName(), data.getType(), data.getTargetId());
                    }
                    log.info("完成处理{}条数据，开始进入下一个循环", count);
                    c = 0;
                } else {
                    var index = Math.min(c, max);
                    var timeout = syncTriggerProperty.getPollingTimeList()[index];
                    log.info("未查询到调整的数据，开始休眠" + timeout + "ms");
                    try {
                        lock.wait(timeout);
                        c++;
                    } catch (InterruptedException e) {
                        log.error("等待异常", e);
                    }
                }
            }
        }
    }

}
